package com.hefei.garden.config.kafka.config;

import com.alibaba.cloud.commons.lang.StringUtils;
import com.google.common.collect.Maps;
import com.hefei.garden.config.kafka.annotation.KafkaTopicStrategy;
import com.hefei.garden.pojo.KafkaBaseDto;
import com.hefei.garden.service.KafkaTopicStrategyService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.context.annotation.ClassPathScanningCandidateComponentProvider;
import org.springframework.core.type.filter.AssignableTypeFilter;

import java.util.Map;
import java.util.Set;

/**
 * @author: devil
 * @Date: 2022/10/7 10
 * @Description:
 */
public class KafkaTopicStrategyContext {

    public static final Map<String, KafkaTopicStrategyService> kafkaTopicStrategyServiceMap = Maps.newConcurrentMap();

    private KafkaTopicStrategyService kafkaTopicStrategyService;

    public KafkaTopicStrategyContext(KafkaBaseDto kafkaBaseDto) {
        String basePackage = "com.hefei.garden";
        // spring提供的方法获取所有 KafkaTopicStrategyService 的子类
        ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(false);
        provider.addIncludeFilter(new AssignableTypeFilter(KafkaTopicStrategyService.class));
        Set<BeanDefinition> beanDefinitionSet = provider.findCandidateComponents(basePackage);
        String curTopic = kafkaBaseDto.getTopic();
        String curGroupId = kafkaBaseDto.getGroupId();
        for (BeanDefinition beanDefinition : beanDefinitionSet) {
            // 获取拦截器类名
            String beanClassName = beanDefinition.getBeanClassName();
            Class<?> aClass = null;
            try {
                aClass = Class.forName(beanClassName);
            } catch (Exception ignored) {
            }
            assert aClass != null;
            // 获取类注解
            KafkaTopicStrategy kafkaTopicStrategy = aClass.getAnnotation(KafkaTopicStrategy.class);
            String topic = kafkaTopicStrategy.topic();
            String groupId = kafkaTopicStrategy.groupId();
            if (!StringUtils.equals(curTopic, topic) || !StringUtils.equals(curGroupId, groupId)) {
                continue;
            }
            try {
                this.kafkaTopicStrategyService = (KafkaTopicStrategyService) aClass.newInstance();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public void businessProcess(ConsumerRecord<String, String> record) {
        kafkaTopicStrategyService.businessProcess(record);
    }

}
